package com.liwy3.consume;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.record.Record;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * Copyright © 元气森林（北京）食品科技集团有限公司. All rights reserved.
 *
 * @Description:
 * @author: Crazy4J
 * @date: 2022年04月21日 14:52
 * @version: V1.0
 */
@Component
@Slf4j
public class Consume2 {


    @KafkaListener(groupId = "topic-test1-group", topics = "my-topic-test1", concurrency = "2")
    public void listen(ConsumerRecord<String, String> record) {
    /*    if (input.equals("fail")) {
            throw new RuntimeException("error...");
        }
        log.info("线程编号:{},input2 value: {}",Thread.currentThread().getId(), input);
     */

        long offset = record.offset();
        String value = record.value();
        int partition = record.partition();
        String topic = record.topic();
        System.out.println("==============================打印kafka日志start================================");

        System.out.println("topic:" + topic + ";offset:+" + offset + ";value:" + value + ";partition:" + partition);

        System.out.println("==============================打印kafka日志end================================");
    }


    @KafkaListener(groupId = "topic-test1-g2", topics = "my-topic-test1")
    public void listen2(ConsumerRecord<String, String> record) {

        log.info("线程编号:{},input2 value: {}", Thread.currentThread().getId(), record.value());


    }

}
